//import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
//import org.apache.flink.streaming.api.watermark.Watermark
//
//@SerialVersionUID(-742759155861320823L)
//class CustomWatermarkExtractor extends AssignerWithPeriodicWatermarks[Nothing] {
//  private var currentTimestamp = Long.MinValue
//
//  override def extractTimestamp(event: Nothing, previousElementTimestamp: Long): Long = { // the inputs are assumed to be of format (message,timestamp)
//    this.currentTimestamp = event.getTimestamp
//    event.getTimestamp
//  }
//
//  @Nullable override def getCurrentWatermark = new Watermark(if (currentTimestamp == Long.MIN_VALUE) Long.MIN_VALUE
//  else currentTimestamp - 1)
//}